package io.renren.modules.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * Mqtt client端
 *
 * @author lhb
 * @date 2023/6/6
 */

@Slf4j
public class MqttPushClient {

    private String host;
    private String clientid;
    private String username;
    private String password;
    private String topic;
    private int timeout;
    private int keepalive;

    @Autowired
    private SampleCallback callback;

    /**
     * 连接配置
     */
    private MqttConnectOptions options;

    public static MqttAsyncClient mqttClient;

    /**
     * 客户端连接
     *
     * @param host
     * @param clientid
     * @param username
     * @param password
     * @param timeout
     * @param topic
     * @param keepalive
     */
    public MqttPushClient(String host, String clientid, String username, String password, String topic, int timeout, int keepalive) {
        this.host = host;
        this.clientid = clientid;
        this.username = username;
        this.password = password;
        this.topic = topic;
        this.timeout = timeout;
        this.keepalive = keepalive;
    }

    /**
     * 连接MQTT服务器
     */
    public synchronized void connect() {
        if (options == null) {
            setOptions();
        }
        if (mqttClient == null) {
            createClient();
        }
        while (!mqttClient.isConnected()) {
            try {
                log.info("开始连接mqtt服务：{}", host);
                IMqttToken token = mqttClient.connect(options);
                token.waitForCompletion();
            } catch (Exception me) {
                log.debug("=====>>>>>mqtt连接失败 message={}", me.getMessage());
                me.printStackTrace();
            }
        }
    }

    /**
     * 创建客户端
     */
    private void createClient() {
        if (mqttClient == null) {
            try {
              /*host为主机名，clientId是连接MQTT的客户端ID，MemoryPersistence设置clientId的保存方式
                默认是以内存方式保存*/
                mqttClient = new MqttAsyncClient(host, clientid, new MemoryPersistence());
                //设置回调函数
                mqttClient.setCallback(callback);
                log.debug("====>>>mqtt客户端启动成功");
            } catch (MqttException e) {
                log.error("mqtt客户端连接错误 error={}", e.getMessage());
                e.printStackTrace();
            }
        }
    }

    /**
     * 断开与mqtt的连接
     */
    public synchronized void disconnect() {
        //判断客户端是否null 是否连接
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                IMqttToken token = mqttClient.disconnect();
                token.waitForCompletion();
            } catch (MqttException e) {
                log.error("====>>>>断开mqtt连接发生错误 message={}", e.getMessage());
            }
        }
        mqttClient = null;
    }

    /**
     * 重新连接MQTT
     */
    public synchronized void refresh() {
        log.debug("====>>>>重新连接MQTT，{}", host);
        disconnect();
        setOptions();
        createClient();
        connect();
    }

    /**
     * 设置连接属性
     */
    private void setOptions() {
        if (options != null) {
            options = null;
        }
        options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        //设置自动重新连接
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        log.debug("====>>>>设置mqtt参数成功");
    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param topic
     * @param pushMessage
     */

    public void pushlish(String topic, String pushMessage) {
        pushlish(0, false, topic, pushMessage);
    }

    /**
     * 发布
     *
     * @param qos         连接方式
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */

    public void pushlish(int qos, boolean retained, String topic, String pushMessage) {
        log.info("mqtt client 发布消息：{}, {}", topic, pushMessage);
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());

        if (StringUtils.isEmpty(topic)) {
            log.error("topic not exist");
            return;
        }
        try {
            IMqttDeliveryToken token = mqttClient.publish(topic, message);
            token.waitForCompletion();
        } catch (MqttException e) {
            e.printStackTrace();
        }

    }

    public String getTopic() {
        return topic;
    }

    public void subscribe(String topic, int qos) {
        try {
            for (String t : topic.split(",")) {
                log.info("开始订阅主题:{}", t);
                mqttClient.subscribe(t, qos);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}
